package com.atguigu.realtime.app.dwd.db;

import com.atguigu.realtime.utils.KafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdTradePaymentInfo {
    public static void main(String[] args) {
        //TODO 1 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //TODO 2 读取Kafka主题数据 封装为Flink表
        tableEnv.executeSql("create table topic_db(\n" +
                "`database` string,\n" +
                "`table` string,\n" +
                "`type` string,\n" +
                "`data` map<string,string>,\n" +
                "`old` map<string,string>,\n" +
                "`ts` string,\n" +
                "`pt` as PROCTIME()\n" +
                ")" + KafkaUtil.getKafkaDDL("topic_db","dwd_trade_payment_info"));

        //TODO 3 读取支付表数据
        Table paymentInfo = tableEnv.sqlQuery("select\n" +
                "`data`['id'] id,\n" +
                "`data`['out_trade_no'] out_trade_no,\n" +
                "`data`['order_id'] order_id,\n" +
                "`data`['alipay_trade_no'] alipay_trade_no,\n" +
                "`data`['total_amount'] total_amount,\n" +
                "`data`['trade_body'] trade_body,\n" +
                "`data`['payment_type'] payment_type,\n" +
                "`data`['payment_status'] payment_status,\n" +
                "`data`['create_time'] create_time,\n" +
                "`data`['update_time'] update_time,\n" +
                "`data`['callback_content'] callback_content,\n" +
                "`data`['callback_time'] callback_time\n" +
                "from\n" +
                "`topic_db`\n" +
                "where\n" +
                "`table` = 'payment_info'\n" +
                "and `type` = 'insert'");
        tableEnv.createTemporaryView("payment_info",paymentInfo);

        //TODO 4 获取订单明细表
        Table resultTable = tableEnv.sqlQuery("select\n" +
                "id,\n" +
                "out_trade_no,\n" +
                "order_id,\n" +
                "alipay_trade_no,\n" +
                "total_amount,\n" +
                "trade_body,\n" +
                "payment_type,\n" +
                "payment_status,\n" +
                "create_time,\n" +
                "update_time,\n" +
                "callback_content,\n" +
                "callback_time\n" +
                "from\n" +
                "payment_info");
        tableEnv.createTemporaryView("result_table",resultTable);

        //TODO 5 建立 Kafka-Connector dwd_trade_payment_info表
        tableEnv.executeSql("create table dwd_trade_payment_info(\n" +
                "id String,\n" +
                "out_trade_no String,\n" +
                "order_id String,\n" +
                "alipay_trade_no String,\n" +
                "total_amount String,\n" +
                "trade_body String,\n" +
                "payment_type String,\n" +
                "payment_status String,\n" +
                "create_time String,\n" +
                "update_time String,\n" +
                "callback_content String,\n" +
                "callback_time String\n" +
                ")" + KafkaUtil.getKafkaSinkDDL("dwd_trade_payment_info"));

        //TODO 6 将关联结果写入 Kafka-Connector 表
        tableEnv.executeSql("insert into dwd_trade_payment_info select * from result_table");
    }
}
